package com.fwmagic.flink.streaming;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StreamingDemoFold {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataSource = env.socketTextStream("localhost", 8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = dataSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = line.split("\\s");
                for (String word : words) {
                    collector.collect(Tuple2.of(word, 1));
                }
            }
        });

        /**
         *  通过fold函数的 第一参数来设置一个初始值，函数第一次运行时会在累加的时候加上初始值
         */
       /* SingleOutputStreamOperator<String> foldData = wordAndOne.keyBy(0).fold("", new FoldFunction<Tuple2<String, Integer>, String>() {
            @Override
            public String fold(String initialValue, Tuple2<String, Integer> tuple2) throws Exception {
                if ("".equals(initialValue)) {
                    return tuple2.f0 + "-" + tuple2.f1;
                } else {
                    String[] arr = initialValue.split("-");
                    tuple2.f1 += Integer.valueOf(arr[1]);
                    return tuple2.f0 + "-" + tuple2.f1;
                }
            }
        });
*/

        /*SingleOutputStreamOperator<Tuple2<String, Integer>> foldData = wordAndOne.keyBy(0).fold(Tuple2.of("", 0), new FoldFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> fold(Tuple2<String, Integer> initialTuple, Tuple2<String, Integer> tuple2) throws Exception {
                String key = tuple2.f0;
                Integer count = tuple2.f1;
                initialTuple.f0 =key;
                initialTuple.f1 += count;
                return initialTuple;
            }
        });*/

        SingleOutputStreamOperator<Tuple2<String, Integer>> foldData = wordAndOne.keyBy(0).sum(1);
        foldData.writeAsCsv("/Users/fangwei/Downloads/bb").setParallelism(1);
        //foldData.print();
        //自定义sink
       /* foldData.addSink(new RichSinkFunction<Tuple2<String,Integer>>() {
            @Override
            public void invoke(Tuple2<String, Integer> value, Context context) throws Exception {
                //获取StreamingRuntimeContext
                StreamingRuntimeContext runtimeContext = (StreamingRuntimeContext) getRuntimeContext();
                //获取Subtask索引值
                int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
                //Subtask最大值
                int numberOfParallelSubtasks = runtimeContext.getNumberOfParallelSubtasks();
                System.err.println("numberOfParallelSubtasks:"+numberOfParallelSubtasks);
                System.out.println(indexOfThisSubtask+">"+value);
            }
        });*/
        env.execute();

    }
}
